package com.example.mq.mqserver;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.common.network.BasicArguments;
import com.example.mq.common.network.BasicReturns;
import com.example.mq.common.network.Request;
import com.example.mq.common.network.Response;
import com.example.mq.common.network.expansion.*;
import com.example.mq.mqserver.VirtualHost;
import com.example.mq.mqserver.core.*;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * BrokerServer 是 消息队列的本体服务器
 * 本质上还是基于 tcp 进行网络传输
 * 通过这个类 实现服务器与客户端交互
 */
@Slf4j
public class BrokerServer {
    //  服务器套接字
    private final ServerSocket serverSocket;
    //  存放与客户端的连接，key=channelId
    private final ConcurrentHashMap<String, Socket> sessionMap = new ConcurrentHashMap<>();
    //  虚拟主机，当前服务器只实现一个虚拟主机
    private final VirtualHost virtualHost = new VirtualHost("defaultHost");
    //  引入线程池，执行服务器与客户端的交互
    private ExecutorService executorService;
    //  引入一个变量表示服务器的运行状态
    private boolean runnable = true;

    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(port);
    }
    public void start() {
        log.info("<-----------------------------服务器启动----------------------------->");
        //  初始化线程池，不指定大小
        executorService = Executors.newCachedThreadPool();
        while (runnable) {
            try {
                Socket clientSocket = serverSocket.accept();
                executorService.submit(() -> {
                    processConnection(clientSocket);
                });

            } catch (SocketException e) {
                //  正常运行
            } catch (IOException e) {
                e.printStackTrace();
                log.info("<-----------------------------服务器关闭----------------------------->");
            }

        }
    }

    //  一般来说，进程停止，服务器就停了
    //  方便测试，手动停止
    public void close() throws IOException {
        runnable = false;
        executorService.shutdownNow();
        serverSocket.close();
        log.info("<-----------------------------服务器关闭----------------------------->");
    }

    //  处理客户端的请求
    private void processConnection(Socket clientSocket) {
        try (InputStream inputStream = clientSocket.getInputStream();
             OutputStream outputStream = clientSocket.getOutputStream()) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream);
                DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                while (true) {
                    //  读请求
                    Request request = readRequest(dataInputStream);
                    //  处理响应
                    Response response = process(request,clientSocket);
                    //  返回响应
                    writeResponse(response,dataOutputStream);
                }
            }

        } catch (EOFException | SocketException e) {
            //  当客户端下线时，要把对应的消费者删除掉
            //  找到consumerTag
            String consumerTag = "";
            for(Map.Entry<String,Socket> entry : sessionMap.entrySet()) {
                if(entry.getValue() == clientSocket) {
                    consumerTag = entry.getKey();
                }
            }
            Consumer consumer = virtualHost.getConsumerManager().getConsumer(consumerTag);
            MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(consumer.getQueueName());
            queue.removeConsumer(consumer);
            virtualHost.getConsumerManager().removeConsumer(consumerTag);

            log.info("客户端下线. " + clientSocket);
        } catch (ClassNotFoundException | MqException | IOException e) {
            log.warn("connection 异常!");
            e.printStackTrace();
        } finally {
            try {
                //  关闭客户端连接
                clientSocket.close();
                //  清理一次连接中的多个channel
                closeClosedSession(clientSocket);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request = new Request();
        request.setType(dataInputStream.readInt());
        request.setLength(dataInputStream.readInt());
        //  读取 payload
        byte[] payload = new byte[request.getLength()];
        int size = dataInputStream.read(payload);
        if(size != request.getLength()) {
            throw new IOException("读取请求异常! ");
        }
        request.setPayload(payload);

        return request;
    }

    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
        byte[] payload = request.getPayload();
        //  先反序列化参数，使用父类引用起来
        BasicArguments basicArguments = (BasicArguments) BinaryTool.fromBytes(payload);

        //  解析 type，执行对应api
        int type = request.getType();
        boolean ok = true;//  表示方法返回值
        if(type == 0x1) {
            //  创建channel
            sessionMap.put(basicArguments.getChannelId(),clientSocket);
            log.info("创建channel成功. channelId=" + basicArguments.getChannelId());
        } else if (type == 0x2) {
            //  销毁channel
            sessionMap.remove(basicArguments.getChannelId());
            log.info("销毁channel成功. channelId=" + basicArguments.getChannelId());
        } else if(type == 0x3) {
            //  创建交换机
            ExchangeDeclareArguments arguments = (ExchangeDeclareArguments) basicArguments;
            ok = virtualHost.exchangeDeclare(arguments.getExchangeName(), arguments.getExchangeType(),
                    arguments.isDurable(), arguments.isAutoDelete(), arguments.getArguments());
        } else if(type == 0x4) {
            //  删除交换机
            ExchangeDeleteArguments arguments = (ExchangeDeleteArguments) basicArguments;
            ok = virtualHost.exchangeDelete(arguments.getExchangeName());
        } else if(type == 0x5) {
            //  创建队列
            QueueDeclareArguments arguments = (QueueDeclareArguments) basicArguments;
            ok = virtualHost.queueDeclare(arguments.getQueueName(),arguments.isDurable(),
                    arguments.isExclusive(),arguments.isAutoDelete(),arguments.getArguments());
        } else if(type == 0x6) {
            //  删除队列
            QueueDeleteArguments arguments = (QueueDeleteArguments) basicArguments;
            ok = virtualHost.queueDelete(arguments.getQueueName());
        } else if(type == 0x7) {
            //  创建绑定
            QueueBindArguments arguments = (QueueBindArguments) basicArguments;
            ok = virtualHost.queueBind(arguments.getExchangeName(),arguments.getQueueName(),arguments.getBindingKey());
        } else if(type == 0x8) {
            //  解除绑定
            QueueUnbindArguments arguments = (QueueUnbindArguments) basicArguments;
            ok = virtualHost.queueUnbind(arguments.getExchangeName(),arguments.getQueueName());
        } else if(type == 0x9) {
            //  发送消息
            BasicPublishArguments arguments = (BasicPublishArguments) basicArguments;
            ok = virtualHost.basicPublish(arguments.getExchangeName(),arguments.getBasicProperties(),
                    arguments.getRoutingKey(),arguments.getBody());
        } else if(type == 0xa) {
            //  订阅队列
            BasicConsumeArguments arguments = (BasicConsumeArguments) basicArguments;
            ok = virtualHost.basicConsume(arguments.getConsumerTag(), arguments.getQueueName(), arguments.isAutoAck(), new Consume() {
                @Override
                public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
                    //  此处的回调就是把收到的消息推送给消费者
                    //  consumerTag 就是 channelId
                    Socket socket = sessionMap.get(consumerTag);
                    if(socket == null || socket.isClosed()) {
                        throw new MqException("消费者已关闭,无法进行消费! consumerTag=" + consumerTag);
                    }

                    //  构造响应
                    //  构造响应的返回值
                    SubscribeReturns subscribeReturns = new SubscribeReturns();
                    subscribeReturns.setChannelId(consumerTag);
                    subscribeReturns.setRid("");//  这个响应是服务器主动推送给客户端，没有对应请求，因此没有必要设置Id
                    subscribeReturns.setOk(true);
                    subscribeReturns.setConsumerTag(consumerTag);
                    subscribeReturns.setBasicProperties(basicProperties);
                    subscribeReturns.setBody(body);
                    //  封装响应
                    Response response = new Response();
                    response.setType(0xc);
                    byte[] payload = BinaryTool.toBytes(subscribeReturns);
                    response.setLength(payload.length);
                    response.setPayload(payload);

                    //  发送响应
                    DataOutputStream dataOutputStream = new DataOutputStream(socket.getOutputStream());
                    writeResponse(response,dataOutputStream);
                    }
            });
        } else if(type == 0xb) {
            //  手动应答
            BasicAckArguments arguments = (BasicAckArguments) basicArguments;
            ok = virtualHost.basicAck(arguments.getQueueName(),arguments.getMessageId());
        } else {
            //  其他非法情况
            throw new MqException("未知的请求! type=" + type);
        }

        //  构造返回信息
        BasicReturns basicReturns = new BasicReturns();
        basicReturns.setRid(basicArguments.getRid());
        basicReturns.setChannelId(basicArguments.getChannelId());
        basicReturns.setOk(ok);
        payload = BinaryTool.toBytes(basicReturns);
        //  封装响应
        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payload.length);
        response.setPayload(payload);
        return response;
    }

    private void writeResponse(Response response, DataOutputStream dataOutputStream) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());
        //  刷新缓冲区
        dataOutputStream.flush();
    }


    private void closeClosedSession(Socket clientSocket) {
        //  注意：不能一边遍历一边删除数据，这样会破坏数据原本的结构
        List<String> deleteChannelIdList = new ArrayList<>();
        for(Map.Entry<String,Socket> entry : sessionMap.entrySet()) {
            if(entry.getValue() == clientSocket) {
                deleteChannelIdList.add(entry.getKey());
            }
        }
        for(String channelId : deleteChannelIdList) {
            sessionMap.remove(channelId);
        }
        log.info("清理session完成! toDeleteChannelIdList=" + deleteChannelIdList);
    }




}
